package cn.virens.common.util.components.queue;

import cn.virens.common.util.exception.APIException;
import cn.virens.common.util.exception.ExceptionUtil;
import cn.virens.common.util.fun.ExFunction;
import org.dromara.hutool.core.thread.ThreadUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

public class VirQueueThread<T> implements AutoCloseable {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final ExecutorService executor;

    private final Integer queueSzie;
    private final Integer queueMax;

    private VirQueueThread(Integer queueSzie, Integer queueMax) {
        this.executor = ThreadUtil.newExecutor(queueSzie);
        this.queueSzie = queueSzie;
        this.queueMax = queueMax;
    }

    /**
     * 创建多线程处理任务
     *
     * @param <T>       返回类型
     * @param queueSzie 最大线程数量
     * @param queueMax  每个线程处理数量
     * @return
     */
    public static <T> VirQueueThread<T> of(Integer queueSzie, Integer queueMax) {
        return new VirQueueThread<>(queueSzie, queueMax);
    }

    /**
     * 创建线程池对列表数据进行处理
     *
     * @param <R>        返回类型
     * @param collection 列表
     * @param function   数据处理(禁止任何异常)
     * @return 响应结果列表
     * @throws Exception
     */
    public <R> List<R> executor(Collection<T> collection, ExFunction<T, R> function) throws Exception {
        if (collection == null || collection.isEmpty() || function == null) {
            return Collections.emptyList();//
        }

        // 根据列表生成队列 & 计算当前需要使用多少线程来执行---------------------------------
        Queue<T> queue = threadQueue(collection.size(), collection);
        int threadSize = threadSize(collection.size(), queueMax);

        // 创建线程池 & 转为队列------------------------------------------------------
        List<Future<List<R>>> futures = new ArrayList<>();

        // 启动多线程进行处理---------------------------------------------------------
        for (int index = 0; index < threadSize; index++) {
            futures.add(executor.submit(() -> {
                T item = null;

                List<R> temp = new ArrayList<>();

                // 循环从待处理队列中处理数据-------------------------------------------
                while ((item = queue.poll()) != null) {
                    temp.add(accept(item, function));
                }

                // 返回当前线程名称
                return temp;
            }));
        }

        List<R> answer = new ArrayList<>();

        // 等待全部执行完成
        for (Future<List<R>> future : futures) {
            answer.addAll(future.get());
        }

        return answer;
    }

    protected <R> R accept(T item, ExFunction<T, R> fun) {
        return ExceptionUtil.safe1(logger, () -> fun.apply(item));
    }

    protected Queue<T> threadQueue(int size, Collection<T> collection) {
        return new ArrayBlockingQueue<>(size, false, collection);
    }

    protected int threadSize(double num1, int num2) throws APIException {
        double result = Math.ceil(num1 / num2);

        if (result < queueSzie) {
            return (int) result;
        } else {
            return queueSzie;
        }
    }

    public void destroy() throws Exception {
        this.executor.shutdownNow();
    }

    @Override
    public void close() throws Exception {
        this.executor.shutdownNow();
    }
}
